#!/usr/bin/env python
# coding: utf-8

"""
钉钉通知实现
"""

import requests
import json
import hmac
import hashlib
import base64
import time
import os
import asyncio
from urllib.parse import quote_plus
from typing import Optional, Dict, Any, List
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

class DingTalkBot:
    def __init__(self, webhook: Optional[str] = None, secret: Optional[str] = None):
        """
        初始化钉钉机器人
        
        参数:
            webhook: 钉钉机器人的webhook地址，默认从环境变量DINGTALK_WEBHOOK获取
            secret: 钉钉机器人的加签密钥，默认从环境变量DINGTALK_SECRET获取
        """
        self.webhook = webhook or os.getenv("DINGTALK_WEBHOOK")
        self.secret = secret or os.getenv("DINGTALK_SECRET")
        
        if not self.webhook or not self.secret:
            raise ValueError("DingTalk webhook and secret must be provided either as parameters or in environment variables.")

    def _get_sign(self) -> str:
        """
        生成签名
        
        返回:
            str: 签名字符串
        """
        if not self.secret:
            return ""
        
        timestamp = str(round(time.time() * 1000))
        string_to_sign = f'{timestamp}\n{self.secret}'
        hmac_code = hmac.new(
            self.secret.encode('utf-8'),
            string_to_sign.encode('utf-8'),
            digestmod=hashlib.sha256
        ).digest()
        sign = quote_plus(base64.b64encode(hmac_code))
        return f"&timestamp={timestamp}&sign={sign}"

    def send_message(self, title: str, content: str, **kwargs) -> Dict[str, Any]:
        """
        发送消息（默认使用markdown格式）
        
        参数:
            title: 消息标题
            content: 消息内容
            **kwargs: 其他参数，包括at_mobiles和is_at_all
            
        返回:
            Dict[str, Any]: 响应结果
        """
        at_mobiles = kwargs.get("at_mobiles", [])
        is_at_all = kwargs.get("is_at_all", False)
        
        return self.send_markdown(title, content, at_mobiles, is_at_all)

    def send_text(self, content: str, at_mobiles: Optional[List[str]] = None, is_at_all: bool = False) -> Dict[str, Any]:
        """
        发送文本消息
        
        参数:
            content: 消息内容
            at_mobiles: 需要@的手机号列表
            is_at_all: 是否@所有人
            
        返回:
            Dict[str, Any]: 响应结果
        """
        data = {
            "msgtype": "text",
            "text": {
                "content": content
            },
            "at": {
                "atMobiles": at_mobiles or [],
                "isAtAll": is_at_all
            }
        }
        return self._send_message(data)

    def send_link(self, title: str, text: str, message_url: str, pic_url: Optional[str] = None) -> Dict[str, Any]:
        """
        发送链接消息
        
        参数:
            title: 消息标题
            text: 消息内容
            message_url: 点击消息跳转的URL
            pic_url: 图片URL（可选）
            
        返回:
            Dict[str, Any]: 响应结果
        """
        data = {
            "msgtype": "link",
            "link": {
                "title": title,
                "text": text,
                "messageUrl": message_url,
                "picUrl": pic_url or ""
            }
        }
        return self._send_message(data)

    def send_markdown(self, title: str, text: str, at_mobiles: Optional[List[str]] = None, is_at_all: bool = False) -> Dict[str, Any]:
        """
        发送markdown消息
        
        参数:
            title: 消息标题
            text: markdown格式的消息内容
            at_mobiles: 需要@的手机号列表
            is_at_all: 是否@所有人
            
        返回:
            Dict[str, Any]: 响应结果
        """
        data = {
            "msgtype": "markdown",
            "markdown": {
                "title": title,
                "text": text
            },
            "at": {
                "atMobiles": at_mobiles or [],
                "isAtAll": is_at_all
            }
        }
        return self._send_message(data)

    def _send_message(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """
        发送消息的底层方法
        
        参数:
            data: 消息数据
            
        返回:
            Dict[str, Any]: 响应结果
        """
        try:
            headers = {'Content-Type': 'application/json'}
            url = f"{self.webhook}{self._get_sign()}"
            response = requests.post(url, headers=headers, data=json.dumps(data))
            return response.json()
        except Exception as e:
            print(f"发送钉钉消息失败: {e}")
            return {"errcode": -1, "errmsg": str(e)} 

# 全局钉钉机器人实例
_dingtalk_bot = None

def get_dingtalk_bot() -> DingTalkBot:
    """
    获取钉钉机器人实例（单例模式）
    
    返回:
        DingTalkBot: 钉钉机器人实例
    """
    global _dingtalk_bot
    if _dingtalk_bot is None:
        try:
            _dingtalk_bot = DingTalkBot()
        except ValueError as e:
            print(f"初始化钉钉机器人失败: {e}")
            # 返回一个空实现，避免程序崩溃
            class NullBot:
                def send_message(self, *args, **kwargs):
                    print("钉钉通知未配置，消息未发送")
                    return {"errcode": -1, "errmsg": "钉钉通知未配置"}
            _dingtalk_bot = NullBot()
    return _dingtalk_bot

async def send_dingtalk_message(title: str, content: str, **kwargs) -> Dict[str, Any]:
    """
    异步发送钉钉消息
    
    参数:
        title: 消息标题
        content: 消息内容
        **kwargs: 其他参数，包括at_mobiles和is_at_all
        
    返回:
        Dict[str, Any]: 响应结果
    """
    # 使用线程池执行同步代码
    loop = asyncio.get_event_loop()
    bot = get_dingtalk_bot()
    
    # 在线程池中执行发送操作
    return await loop.run_in_executor(
        None, 
        lambda: bot.send_message(title, content, **kwargs)
    ) 